﻿using FlyingEye.Monitors.Repositories;
using FlyingEye.SignalRHubs;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging;
using Volo.Abp.Application.Services;

namespace FlyingEye.DeviceMonitor
{
    /// <summary>
    /// 消息派遣
    /// </summary>
    public class MessageDispatcherService : ApplicationService
    {
        private readonly IHubContext<MonitorServiceHub, IMessageDispatcherClient> _hubContext;
        private readonly ILogger<MessageDispatcherService> _logger;

        private readonly SignalRClientMapping _signalRClientMapping;
        private readonly SignalRGroupMapping _signalRGroupMapping;

        private readonly IStringValueMonitorClientRepository _stringValueMonitorClientRepository;
        private readonly IStringValueMonitorGroupRepository _stringValueMonitorGroupRepository;

        private readonly INumericValueMonitorClientRepository _nmericValueMonitorClientRepository;
        private readonly INumericValueMonitorGroupRepository _numericValueMonitorGroupRepository;

        public MessageDispatcherService(
            IHubContext<MonitorServiceHub, IMessageDispatcherClient> hubContext,
            ILogger<MessageDispatcherService> logger,
            SignalRClientMapping signalRClientMapping,
            SignalRGroupMapping signalRGroupMapping,
            IStringValueMonitorClientRepository stringValueMonitorClientRepository,
            IStringValueMonitorGroupRepository stringValueMonitorGroupRepository,
            INumericValueMonitorClientRepository nmericValueMonitorClientRepository,
            INumericValueMonitorGroupRepository numericValueMonitorGroupRepository)
        {
            _hubContext = hubContext;
            _logger = logger;
            _signalRClientMapping = signalRClientMapping;
            _signalRGroupMapping = signalRGroupMapping;
            _stringValueMonitorClientRepository = stringValueMonitorClientRepository;
            _stringValueMonitorGroupRepository = stringValueMonitorGroupRepository;
            _nmericValueMonitorClientRepository = nmericValueMonitorClientRepository;
            _numericValueMonitorGroupRepository = numericValueMonitorGroupRepository;
        }

        public async Task DispatcherAsync(DispatcherMessage message)
        {
            await DispatcherNumericValueMessageToClientAsync(message);
            await DispatcherNumericValueMessageToGroupAsync(message);
            await DispatcherStringValueMessageToClientAsync(message);
            await DispatcherStringValueMessageToGroupAsync(message);
        }

        private async Task DispatcherNumericValueMessageToClientAsync(DispatcherMessage message)
        {
            int pageIndex = 0; // 假设从 0 开始
            const int pageSize = 100;
            var parallelOptions = new ParallelOptions();
            parallelOptions.MaxDegreeOfParallelism = 25; // 并行度
            var query = await _nmericValueMonitorClientRepository.GetQueryableAsync();

            while (true)
            {
                var clientMonitorPoints = query.OrderBy(e => e.Id)
                     .Where(t => t.ReadNumericValueTaskId == message.Task.Id)
                     .Skip(pageIndex * pageSize)
                .Take(pageSize);

                if (!clientMonitorPoints.Any())
                {
                    break;
                }

                await Parallel.ForEachAsync(clientMonitorPoints, parallelOptions, async (point, token) =>
                {
                    try
                    {
                        if (_signalRClientMapping.TryGetConnectionId(point.ClientName, out var connectionId))
                        {
                            await _hubContext.Clients.Client(connectionId)
                                .OnDispatcherAsync(message);
                        }
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, $"{point.Id} 监视点通知失败！");
                    }
                });

                // 更新分页索引
                pageIndex++;
            }
        }

        private async Task DispatcherNumericValueMessageToGroupAsync(DispatcherMessage message)
        {
            int pageIndex = 0; // 假设从 0 开始
            const int pageSize = 100;
            var parallelOptions = new ParallelOptions();
            parallelOptions.MaxDegreeOfParallelism = 25; // 并行度
            var query = await _numericValueMonitorGroupRepository.GetQueryableAsync();

            while (true)
            {
                var groupMonitorPoints = query.OrderBy(e => e.Id)
                     .Where(t => t.ReadNumericValueTaskId == message.Task.Id)
                     .Skip(pageIndex * pageSize)
                .Take(pageSize);

                if (!groupMonitorPoints.Any())
                {
                    break;
                }

                await Parallel.ForEachAsync(groupMonitorPoints, parallelOptions, async (point, token) =>
                {
                    try
                    {
                        if (_signalRGroupMapping.HasGroup(point.GroupName))
                        {
                            await _hubContext.Clients.Group(point.GroupName)
                            .OnDispatcherAsync(message);
                        }
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, $"{point.Id} 监视点通知失败！");
                    }
                });

                // 更新分页索引
                pageIndex++;
            }
        }

        private async Task DispatcherStringValueMessageToGroupAsync(DispatcherMessage message)
        {
            int pageIndex = 0; // 假设从 0 开始
            const int pageSize = 100;
            var parallelOptions = new ParallelOptions();
            parallelOptions.MaxDegreeOfParallelism = 25; // 并行度
            var query = await _stringValueMonitorGroupRepository.GetQueryableAsync();

            while (true)
            {
                var groupMonitorPoints = query.OrderBy(e => e.Id)
                     .Where(t => t.ReadStringValueTaskId == message.Task.Id)
                     .Skip(pageIndex * pageSize)
                .Take(pageSize);

                if (!groupMonitorPoints.Any())
                {
                    break;
                }

                await Parallel.ForEachAsync(groupMonitorPoints, parallelOptions, async (point, token) =>
                {
                    try
                    {
                        if (_signalRGroupMapping.HasGroup(point.GroupName))
                        {
                            await _hubContext.Clients.Group(point.GroupName)
                           .OnDispatcherAsync(message);
                        }
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, $"{point.Id} 监视点通知失败！");
                    }
                });

                // 更新分页索引
                pageIndex++;
            }
        }

        private async Task DispatcherStringValueMessageToClientAsync(DispatcherMessage message)
        {
            int pageIndex = 0; // 假设从 0 开始
            const int pageSize = 100;
            var parallelOptions = new ParallelOptions();
            parallelOptions.MaxDegreeOfParallelism = 25; // 并行度
            var query = await _stringValueMonitorClientRepository.GetQueryableAsync();

            while (true)
            {
                var clientMonitorPoints = query.OrderBy(e => e.Id)
                     .Where(t => t.ReadStringValueTaskId == message.Task.Id)
                     .Skip(pageIndex * pageSize)
                .Take(pageSize);

                if (!clientMonitorPoints.Any())
                {
                    break;
                }

                await Parallel.ForEachAsync(clientMonitorPoints, parallelOptions, async (point, token) =>
                {
                    try
                    {
                        if (_signalRClientMapping.TryGetConnectionId(point.ClientName, out var connectionId))
                        {
                            await _hubContext.Clients.Client(connectionId)
                                .OnDispatcherAsync(message);
                        }
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, $"{point.Id} 监视点通知失败！");
                    }
                });

                // 更新分页索引
                pageIndex++;
            }
        }
    }
}
